//package Autodesk;
//
//import java.io.IOException;
//import java.util.ArrayList;
//import java.util.Hashtable;
//import java.util.List;
//
//import org.apache.hadoop.conf.Configuration;
//import org.apache.hadoop.conf.Configured;
//import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.io.IntWritable;
//import org.apache.hadoop.io.Text;
//import org.apache.hadoop.mapred.FileInputFormat;
//import org.apache.hadoop.mapred.FileOutputFormat;
//import org.apache.hadoop.mapred.JobClient;
//import org.apache.hadoop.mapred.JobConf;
//import org.apache.hadoop.mapred.OutputCollector;
//import org.apache.hadoop.mapred.Reporter;
//import org.apache.hadoop.mapred.TextInputFormat;
//import org.apache.hadoop.mapred.TextOutputFormat;
//import org.apache.hadoop.mapred.lib.ChainMapper;
//import org.apache.hadoop.mapreduce.Mapper;
//import org.apache.hadoop.mapreduce.Reducer;
//import org.apache.hadoop.util.Tool;
//import org.apache.hadoop.util.ToolRunner;
//
//public class SessionFilterComplete  extends Configured implements Tool {
//	public Path prodPath;
//	public static class LogMapper extends Mapper<Object, Text, Text, Text> {
//		
//		// private final static IntWritable one = new IntWritable(1);
//		private Text word = new Text();
//		Hashtable<String, String> ht = new Hashtable<String, String>();
//	//public void map(Object key, Text value, Context context)
//	public void map(Object key, Text value, OutputCollector<Text, Text> output,Reporter reporter)
//			throws IOException, InterruptedException {
//
//		String record[] = value.toString().toLowerCase().split("\t", 12);
//		StringBuffer bf = new StringBuffer();
//		StringBuffer bv=new StringBuffer();
//		try{
//		bf.append(record[0]);// DATE
//		bf.append("\t");
//		bf.append(record[3]);//ipPort
//		word.set(bf.toString());
//		
//		String out=record[7];
//		try{
//			String t1[]=out.split(":");
//			Integer.parseInt(t1[0]);
//		}catch(Exception e){
//			out=record[8];
//			System.out.println("..........................................CHANGE:"+record[7]+"BY"+ out);
//		}
//		
//		bv.append(out);
//		bv.append("\t");
//		bv.append(record[1]);// Channel
//		bv.append("\t");
//		bv.append(record[6]);// Language
//		bv.append("\t");
//		bv.append(record[5]);// Product
//		Text outText=new Text();
//		outText.set(bv.toString());
//		
//		output.collect(word, outText);
//		}catch(Exception e){
//			System.out.println("Error Reg: "+value);
//		}
//		}
//}
//
//	public static class LogReducer extends
//			Reducer<Text, Text, Text, IntWritable> {
//		private int sessioncont=5;
//		private IntWritable result = new IntWritable(1);
//		
//		public void setup(Context context){
//			Configuration conf=context.getConfiguration();
//			      String paramNotReady=conf.get("input.parameter");
//			      sessioncont=Integer.parseInt(paramNotReady);
//		}
//		public void reduce(Text key, Iterable<Text> values,
//				Context context) throws IOException, InterruptedException {
//			ReduceHours red=new ReduceHours();
//			ArrayList<String> listHours=new ArrayList<String>();
//			listHours=red.generateListHours(values);	
//			red.orderList(listHours);
//			sessioncont=sessioncont+1;
//			//System.out.println(key.toString());
//			int i;
//			for ( i=0; i<listHours.size()-1;++i){
//				String line1[]=listHours.get(i).split("\t");
//		    	String line2[]=listHours.get(i+1).split("\t");
//				//System.out.println("----"+line1[0]+"-----"+line2[0]);
//				if(!red.isSameSessionbyHour(line1[0], line2[0], 1)){
//					String newkey=key.toString()+"\t"+line1[1]+"\t"+line1[2]+"\t"+line1[3]+"\t"+sessioncont;
//					context.write(new Text(newkey), result);
//					
//					//System.out.println("--------change session--------");
//					sessioncont=sessioncont+1;
//				}else{
//					String newkey=key.toString()+"\t"+line1[1]+"\t"+line1[2]+"\t"+line1[3]+"\t"+sessioncont;
//					context.write(new Text(newkey), result);
//				}
//			}
//			if(i==listHours.size()-1){
//		    	String line2[]=listHours.get(i).split("\t");
//				String newkey=key.toString()+"\t"+line2[1]+"\t"+line2[2]+"\t"+line2[3]+"\t"+sessioncont;
//				context.write(new Text(newkey), result);
//			}
//		
//		}
//	}
//		
//		public static class LogMapperSort extends Mapper<Object, Text, Text, IntWritable> {
//			
//			// private final static IntWritable one = new IntWritable(1);
//			private Text word = new Text();
//			private IntWritable one= new IntWritable(1);
//			Hashtable<String, String> ht = new Hashtable<String, String>();
//		public void map(Object key, Text value, Context context)
//				throws IOException, InterruptedException {
//
//			String record[] = value.toString().split("\t", 7);
//			StringBuffer bf = new StringBuffer();
//			bf.append(record[0]);// TimeDimID
//			bf.append("\t");
//			bf.append(record[3]);// LanguageID
//			bf.append("\t");
//			bf.append(record[4]);// ProductID
//			bf.append("\t");
//			bf.append(record[2]);// ChannelID
//			bf.append("\t");
//			bf.append(record[5]);// SessionID
//			bf.append("\t");
//			bf.append(record[1]);// IP
//			word.set(bf.toString());
//
//			context.write(word, one);
//		}
//	}
//
//
//
//	public static class LogReducerCounter extends
//		Reducer<Text, IntWritable, Text, IntWritable> {
//
//		private IntWritable result = new IntWritable();
//
//		public void reduce(Text key, Iterable<IntWritable> values,
//			Context context) throws IOException, InterruptedException {
//		
//		int sum = 0;
//		for (IntWritable val : values) {
//			sum += val.get();
//		}
//		result.set(sum);
//		context.write(key, result);
//
//		}
//	}
//	public int run(String[] args) throws Exception { 
//		Configuration conf = getConf();
//		JobConf job=new JobConf(conf);
//		
//		job.setJobName("SessionFilterComplete");
//		job.setInputFormat(TextInputFormat.class);
//		job.setOutputFormat(TextOutputFormat.class);
//		
//		//Job job = new Job(conf, "SessionFilter");
////		job.setJarByClass(Autodesk.SessionFilter.class);
////
////		job.setMapperClass(LogMapper.class);
////		job.setReducerClass(LogReducer.class);
////		
////		job.setMapOutputKeyClass(Text.class);
////		job.setMapOutputValueClass(Text.class);
////		
////		job.setOutputKeyClass(Text.class);
////		job.setOutputValueClass(IntWritable.class);
////
//		List<String> other_args = new ArrayList<String>();
//
//		for (int i = 0; i < args.length; ++i) {
//			if ("-param".equals(args[i])) {
//				job.set("input.parameter", new String(args[++i]));
//			}else{
//				other_args.add(args[i]);
//			}
//		}
//	    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
//	    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
//
//	    JobConf map1Conf=new JobConf(conf);
////	    ChainMapper.addMapper(job,LogMapper.class, Object.class, Text.class, Text.class, Text.class, true, map1Conf);
////	    JobConf reduce1Conf=new JobConf(false);
////	    ChainReducer.setReducer(job, LogReducer.class, Text.class, Text.class, Text.class, Text.class, true, reduce1Conf);
////	    JobConf map2Conf=new JobConf(false);
////	    ChainMapper.addMapper(job,LogMapperSort.class, Object.class, Text.class, Text.class, IntWritable.class, true, map2Conf);
////	    JobConf reduce2Conf=new JobConf(false);
////	    ChainReducer.setReducer(job, LogReducerCounter.class, Text.class, IntWritable.class, Text.class, IntWritable.class, true, reduce1Conf);
//	    JobClient.runJob(job);
//		return 0;
//	
//	}
//	
//	public static void main(String[] args) throws Exception {
//		int res = ToolRunner.run(new Configuration(), new SessionFilterComplete(), args);
//		System.exit(res);
//		}
//}
